package com.doit.day05

import java.sql.{Connection, DriverManager, PreparedStatement}
import scala.io.Source


object _02_导入数据到mysql {
  def main(args: Array[String]): Unit = {

    val movie_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val rating_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val user_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList

    /*
       ETL
     */
    val movies: Map[String, (String, String)] = movie_line.map(line => {
      val arr: Array[String] = line.split("_")
      //电影id    电影名称   电影类型
      (arr(0), (arr(1), arr(2)))
    }).toMap

    val ratings: List[(String, String, String, String)] = rating_line.map(line => {
      val arr: Array[String] = line.split("_")
      //user_id   电影id   评分   时间戳
      (arr(0), arr(1), arr(2), arr(3))
    })

    val users: Map[String, (String, String, String, String)] = user_line.map(line => {
      val arr: Array[String] = line.split("_")
      //user_id  性别   年龄  职业   邮编
      (arr(0), (arr(1), arr(2), arr(3), arr(4)))
    }).toMap

    val conn: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456")
    val pps: PreparedStatement = conn.prepareStatement("insert into ratings values(?,?,?,?)")
    val pps1: PreparedStatement = conn.prepareStatement("insert into users values(?,?,?,?,?)")
    val pps2: PreparedStatement = conn.prepareStatement("insert into movies values(?,?,?)")

    var cnt = 0
/*
    ratings.foreach(tp=>{
      pps.setString(1,tp._1)
      pps.setString(2,tp._2)
      pps.setString(3,tp._3)
      pps.setString(4,tp._4)
      pps.addBatch()
      cnt += 1
      if(cnt % 10000 ==0){
        pps.executeBatch()
        cnt = 0
      }
    }
    )
    pps.executeBatch()*/

    cnt = 0
    movies.foreach(tp=>{
      pps2.setString(1,tp._1)
      pps2.setString(2,tp._2._1)
      pps2.setString(3,tp._2._2)
      pps2.addBatch()
      cnt += 1
      if(cnt %1000 == 0){
        pps2.executeBatch()
        cnt = 0
      }
    }
    )

    pps2.executeBatch()

    cnt = 0
    users.foreach(tp=>{
      pps1.setString(1,tp._1)
      pps1.setString(2,tp._2._1)
      pps1.setString(3,tp._2._2)
      pps1.setString(4,tp._2._3)
      pps1.setString(5,tp._2._4)

      pps1.addBatch()
      cnt += 1
      if (cnt %1000 == 0){
        pps1.executeBatch()
      }
    }
    )
    pps1.executeBatch()

  }

}
